package com.leilei.join.window;

import com.leilei.join.common.Vehicle;
import org.apache.flink.api.common.eventtime.*;

//自定义水位线依据时间 这里采用系统时间
public class VehicleWatermark implements WatermarkStrategy<Vehicle> {

    @Override
    public TimestampAssigner<Vehicle> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
        return (element, recordTimestamp) -> System.currentTimeMillis();
    }

    @Override
    public WatermarkGenerator<Vehicle> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
        return new WatermarkGenerator<Vehicle>() {
            @Override
            public void onEvent(Vehicle event, long eventTimestamp, WatermarkOutput output) {
                output.emitWatermark(new Watermark(System.currentTimeMillis()));
            }

            @Override
            public void onPeriodicEmit(WatermarkOutput output) {
                output.emitWatermark(new Watermark(System.currentTimeMillis()));
            }
        };
    }
}